package dao

import (
	"context"
	"fmt"
	"wxpay/internal/model"
	"wxpay/pkg/mcontext"
	"wxpay/pkg/merrors"
	"wxpay/pkg/util"

	"github.com/nsqio/go-nsq"

	"go.uber.org/zap"
	"gorm.io/gorm"
)

type EventMessage interface {
	// 新建消息
	SaveMessageInTx(ctx context.Context, tx *gorm.DB, msg *model.Message) (err error)
	SaveMessage(ctx context.Context, msg *model.Message) (err error)
	// 更新消息发送状态
	UpdateMessageStatus(ctx context.Context, msgId int64, baseStatus string, targetStatus string) (err error)
	// 根据消息id加载消息
	GetMessageById(ctx context.Context, msgId int64) (msg *model.Message, err error)
	// 发送消息
	PublishMessage(ctx context.Context, msg *model.Message) (err error)
	// 批量发送消息
	MultiPublishMessage(ctx context.Context, topic string, msgs [][]byte) (err error)
	// 加载待发送消息
	GetMessagesAwaitPublish(ctx context.Context, pagination model.Pagination) ([]*model.Message, error)
}

type MessageRepo struct {
	*gorm.DB
	producer *nsq.Producer
}

func NewMessageRepo(d *gorm.DB, p *nsq.Producer) *MessageRepo {
	return &MessageRepo{
		DB:       d,
		producer: p,
	}
}

func (d *MessageRepo) GetLockRepo() *LockRepo {
	return &LockRepo{d.DB}
}

func (d *MessageRepo) SaveMessage(ctx context.Context, msg *model.Message) (err error) {
	db := d.DB.WithContext(ctx)
	if e := db.Transaction(func(tx *gorm.DB) error {
		if err = d.SaveMessageInTx(ctx, tx, msg); err != nil {
			return err
		}
		return nil
	}); e != nil {
		return e
	}
	return nil
}

func (d *MessageRepo) SaveMessageInTx(_ context.Context, tx *gorm.DB, msg *model.Message) (err error) {
	//	保存消息
	const op merrors.Op = "dao.SaveMessageInTx"
	if er := tx.Model(&model.Message{}).Save(msg).Error; er != nil {
		return merrors.E(op, merrors.DBError, er, merrors.ErrInternal)
	}
	return nil
}

func (d *MessageRepo) UpdateMessageStatus(ctx context.Context, msgId int64, baseStatus string, targetStatus string) (err error) {
	//	更新消息发送状态
	const op merrors.Op = "dao.UpdateMessageStatus"
	db := d.DB.WithContext(ctx)

	if e := db.Transaction(func(tx *gorm.DB) error {
		msg, er := d.GetLockRepo().LockMessage(ctx, tx, msgId)
		if er != nil {
			return er
		}
		if msg.Status != baseStatus {
			return merrors.E(op, merrors.StatusInvalid, fmt.Sprintf("message状态:%s，期望:%s", msg.Status, baseStatus), merrors.ErrStatusInValid)
		}
		if er = tx.Model(&model.Message{}).Where("id = ?", msgId).Updates(map[string]interface{}{"status": targetStatus, "update_time": util.GetNow()}).Error; er != nil {
			return merrors.E(op, merrors.DBError, fmt.Sprintf("保存messgae状态失败:%v", er), merrors.ErrInternal)
		}
		switch targetStatus {
		case model.MessagePublishingStatus:
			if er = tx.Model(&model.Message{}).Where("id = ?", msgId).Updates(map[string]interface{}{"retry_count": msg.RetryCount + 1}).Error; er != nil {
				return merrors.E(op, merrors.DBError, fmt.Sprintf("保存messgae retry_count失败:%v", er), merrors.ErrInternal)
			}
		case model.MessageCreatedStatus:
			if er = tx.Model(&model.Message{}).Where("id = ?", msgId).Updates(map[string]interface{}{"eta": msg.GetNextETA()}).Error; er != nil {
				return merrors.E(op, merrors.DBError, fmt.Sprintf("保存messgae eta失败:%v", er), merrors.ErrInternal)
			}
		}
		return nil
	}); e != nil {
		return e
	}
	return nil
}

func (d *MessageRepo) GetMessageById(ctx context.Context, msgId int64) (msg *model.Message, err error) {
	// 根据消息id加载消息
	// const op merrors.Op = "dao.GetMessageById"
	db := d.DB.WithContext(ctx)
	msg = &model.Message{}
	if err := db.Model(&model.Message{}).Where("id = ?", msgId).First(msg).Error; err != nil {
		return nil, err
	}
	return msg, nil
}

func (d *MessageRepo) PublishMessage(ctx context.Context, msg *model.Message) (err error) {
	const op merrors.Op = "dao.PublishMessage"
	l := mcontext.Logger(ctx)
	msg.RetryCount += 1
	if err = d.producer.Publish(msg.Topic, msg.Content); err != nil {
		l.Warn("发送消息失败", zap.String("topic", msg.Topic), zap.Error(err))
		return merrors.E(op, merrors.EventSendFailed, err, merrors.ErrMessageSent)
	}
	l.Info("发送消息成功", zap.String("topic", msg.Topic), zap.ByteString("body", msg.Content))
	return nil
}

// 批量发送消息
func (d *MessageRepo) MultiPublishMessage(ctx context.Context, topic string, msgs [][]byte) (err error) {
	const op merrors.Op = "dao.MultiPublishMessage"
	l := mcontext.Logger(ctx)

	if err = d.producer.MultiPublish(topic, msgs); err != nil {
		l.Warn("批量发送消息失败", zap.String("topic", topic), zap.Error(err))
		return merrors.E(op, merrors.EventSendFailed, err, merrors.ErrMessageSent)
	}
	l.Info("批量发送消息成功", zap.String("topic", topic), zap.ByteStrings("body", msgs))
	return nil
}

func (d *MessageRepo) GetMessagesAwaitPublish(ctx context.Context, pagination model.Pagination) ([]*model.Message, error) {
	// const op merrors.Op = "dao.GetMessagesAwaitPublish"
	db := d.DB.WithContext(ctx)
	var r []*model.Message
	if err := db.Model(&model.Message{}).Where("status = ? and retry_count < ? and eta < ?", model.MessageCreatedStatus, model.MaxRetryCount, util.GetNow()).Order("eta asc").Find(&r).Error; err != nil {
		return nil, err
	}
	return r, nil

}
